package yz.mr.Filtration1;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 需求1：过滤出大于总分450分以上的同学
 *
 * 分析
 *      Mapper阶段：
 *          ①读取ReduceJoin阶段的数据
 *          ②对结果数据进行切分，并对数据进行截取出来
 *          ③对大于450分的同学进行写出
 *
 *      Reduce：
 *          由于处理过程中，不需要对数据进行汇总处理，所以可以不使用该部分代码
 *
 *
 * 对应MapReduce计算时，可以不使用Reduce阶段，不过需要注意的是在main函数需要去除setReduceClass，最终输出的数据和Mapper端输出的数据类型一致
 *
 */
public class FiltrationText1 {
    public static void main(String[] args) throws Exception{

        //1.创建Job操作对象
        //2.创建配置类对象
        Configuration entries = new Configuration();
        Job job = Job.getInstance(entries);

        //3.配置job
        job.setJobName("FiltrationText1");
        job.setJarByClass(FiltrationText1.class);

        //4.设置Mapper以及输出类型
        job.setMapperClass(FiltrationText1Mapper.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);

        //6.设置最终地输出数据类型
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);

        //7.设置输入输出路径(这是本地的路径)  输入路径既可以指定路径也可以指定目录
        FileInputFormat.addInputPath(job,new Path("output/ReduceCount"));
        FileOutputFormat.setOutputPath(job,new Path("output/FiltrationCount"));

        //8.提交执行当前JOB
        job.waitForCompletion(true);

    }
}
